package ucsb.appscale.protoBuff;

import java.util.Map;
import java.util.HashMap;
import java.io.IOException;
import java.util.Iterator;
import java.util.Date;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataByteArray;
import com.google.appengine.api.datastore.*;
import com.google.storage.onestore.v3.OnestoreEntity;
import com.google.apphosting.api.ApiProxy;

public class ParseWithKey extends EvalFunc<Map<String, Object>>{

        public Map<String, Object> exec(Tuple input)throws IOException{
		ApiProxy.setEnvironmentForCurrentThread(new AppEnvironment());
                if (input == null || input.size() == 0)
                        return null;

		Map<String, Object> result = new HashMap<String, Object>();
                byte[] pb_bytes = null;
                try{
                	Object key = input.get(0);
			Object inputValue = input.get(1);
			Tuple tuple = null;
			/*
			System.out.println("Key: " + key.getClass().getName());
			System.out.println("value: " + inputValue.getClass().getName());
			*/
			if((key instanceof String) && (inputValue instanceof DataBag)){ 
				int index = ((String)key).lastIndexOf(':');
				String keyId = ((String)key).substring(index+1);
				char[] keyIdChars = keyId.toCharArray();
				int i;
				for(i=0; i<keyId.length() && keyIdChars[i]=='0'; i++)
				result.put("key", keyId.substring(i+1));
				
				for(Iterator<Tuple> it = ((DataBag)inputValue).iterator(); it.hasNext();){
					tuple = it.next();
				        /*	
					System.out.println("tuple size: " + tuple.size());
					System.out.println("tuple 1: " + tuple.get(0));
					System.out.println("name type: " + tuple.getType(0));
					System.out.println("tuple 2: " + tuple.get(1));
					*/
					String name = tuple.get(0).toString();
					if("Encoded_Entity".equals(name)){
						pb_bytes = ((DataByteArray)tuple.get(1)).get();
					}
				}
			}
			if(pb_bytes != null){
				OnestoreEntity.EntityProto entityProto = new OnestoreEntity.EntityProto();
                		entityProto.parseFrom(pb_bytes);
        			Entity entity = EntityTranslator.createFromPb(entityProto);
				for (Map.Entry<String, Object> en : entity.getProperties().entrySet()) {
					Object value = en.getValue();
					/*
					if(value == null)
						System.out.println("null");
					else
						System.out.println(value.getClass().getName());
					*/
					if(value instanceof Date)
						result.put(en.getKey(), ((Date)value).getTime());
					else if(value instanceof Double)
						result.put(en.getKey(), value);
					else if(value instanceof Key){
						/*						
						System.out.println("key id: " + Long.toString(((Key)value).getId()));
						System.out.println("key kind: " + ((Key)value).getKind());
						System.out.println("key name: " + ((Key)value).getName());
						System.out.println("key namespace: " + ((Key)value).getNamespace());
						System.out.println("key parent: " + ((Key)value).getParent());
						*/
						result.put(en.getKey(), Long.toString(((Key)value).getId()));
					}
					else
						result.put(en.getKey(), value.toString());
				}
			}
		}catch(Exception e){
                        e.printStackTrace();
                }

		return result;
        }

        //For testing
	public static void main(String args[]){
	
	}

}




